#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "shmem.h"
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "msg_queue_peer.h"
#include <sqlite3.h>
#include <pthread.h>
#include <time.h>

#define CLIENTID "ExampleClientPub"
#define TOPIC "MQTT Examples"
#define PAYLOAD "Hello World!"
#define QOS 1
#define TIMEOUT 10000L
#define shmemwhere "/tmp/ipc/shmem/%s"
#define shmemname "shmem"
#define msgqueuewhere "/tmp/ipc/msgqueue/peer/"
#define msgqueuename "msgqueue"
#define pointtablewhere "./node.json"
#define sqlite3dbwhere "./val.db"
volatile int type6;
// 消息队列 共享内存  数据库 点表
sqlite3 *db;
char *errmsg;
union val_t
{

    int bool;
    int a;
    float b;
};

struct std_node
{
    int key;      // 唯一键值
    int type;     // 数据点类型
    int type_dev; // 数据点属于哪个设备，根据网关支持的设备自行定义
    union val_t old_val;
    union val_t new_val;
    int ret; // 默认为-1，采集成功后设置为0，采集失败再置-1
};

struct msgqueue
{
    long msgtype;
    int typekey;
    char data[64];
};

MQTTClient client;
MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
int rc;

volatile MQTTClient_deliveryToken deliveredtoken;
void publicmqtt(char *p)
{
    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;

    MQTTClient_publishMessage(client, "up", &pubmsg, &token);
}

void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;

    payloadptr = message->payload;
    // 接收mqtt
    printf("%s\n", payloadptr);

    cJSON *root = cJSON_Parse(payloadptr);
    cJSON *item = cJSON_GetObjectItem(root, "type");
    int type = item->valueint;

    struct shm_param shmpam;
    shm_init(&shmpam, shmemname, 1024);
    int *p = shm_getaddr(&shmpam);
    struct std_node *dev1 = (struct std_node *)(p + 1);
    struct std_node *dev = dev1;
    if (type == 1)
    {
        // 单次返回数据
        // 序列化
        cJSON *root = cJSON_CreateObject();
        cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
        cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
        cJSON *data = cJSON_CreateArray();
        cJSON_AddItemToObject(root, "data", data);
        cJSON *item;
        for (int i = 0; i < *p; i++)
        {
            item = cJSON_CreateObject();
            cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(dev->key));
            char val[64] = {0};
            if (dev->type == 1)
            {
                sprintf(val, "%d", dev->new_val.bool);
            }
            else if (dev->type == 2)
            {
                sprintf(val, "%d", dev->new_val.a);
            }
            else if (dev->type == 3)
            {
                sprintf(val, "%f", dev->new_val.b);
            }
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
            cJSON_AddItemToArray(data, item);
            dev++;
        }

        publicmqtt(cJSON_Print(root));
    }
    else if (type == 2) // 控制功能
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        item = cJSON_GetObjectItem(data, "key");
        int key = item->valueint;
        item = cJSON_GetObjectItem(data, "val");
        char val[64] = {0};
        strcpy(val, item->valuestring);

        // 写到消息队列
        struct msgqueue recvque;
        if (key >= 301 && key <= 308)
        {
            recvque.msgtype = 3;
        }
        else
        {
            recvque.msgtype = 1;
        }
        recvque.typekey = key;
        printf("%s\n", val);
        strcpy(recvque.data, val);
        msg_queue_send(msgqueuename, &recvque, sizeof(struct msgqueue), 0);
    }
    else if (type == 3) // 模式切换
    {
        cJSON *data = cJSON_GetObjectItem(root, "data");
        item = cJSON_GetObjectItem(data, "type");
        int type2 = item->valueint;
        printf("type2 = %d\n", type2);
        type6 = type2;

        int fd = open(pointtablewhere, O_RDWR); // 打开文件用于读写
        if (fd == -1)
        {
            perror("Failed to open file");
            return -1;
        }

        char *buf1 = (char *)malloc(4096);
        if (buf1 == NULL)
        {
            perror("Memory allocation failed");
            close(fd);
            return -1;
        }

        int len = read(fd, buf1, 4096);
        lseek(fd, 0, SEEK_SET); // 将文件指针移到文件开头
        cJSON *root1 = cJSON_Parse(buf1);
        cJSON *report = cJSON_GetObjectItem(root1, "report");

        // 确保报文类型和更新
        if (type2 == 0)
        { // 主动采集
            cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(0));
        }
        else if (type2 == 1)
        { // 变化上报
            cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(1));
        }
        else if (type2 == 2)
        { // 周期上报
            item = cJSON_GetObjectItem(data, "period");
            int period = item->valueint;
            cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(2));
            cJSON_ReplaceItemInObject(report, "period", cJSON_CreateNumber(period));
        }

        // 获取最终 JSON 字符串并写入文件
        char *json_str = cJSON_Print(root1);
        if (json_str == NULL)
        {
            fprintf(stderr, "Failed to print JSON\n");
            free(buf1);
            close(fd);
            return -1;
        }

        // 截断文件并写入新的数据
        ftruncate(fd, 0);       // 截断文件，删除原有内容
        lseek(fd, 0, SEEK_SET); // 将文件指针移动到文件开始位置

        // 写入新的 JSON 数据
        write(fd, json_str, strlen(json_str));

        // 释放内存
        free(json_str);
        free(buf1);

        close(fd);
    }
    else if (type == 4) // 历史查询
    {

        char *errmsg;
        char **resultp;
        int n_row, n_cloum;
        int i, j;

        cJSON *data = cJSON_GetObjectItem(root, "data");
        item = cJSON_GetObjectItem(data, "key");
        int key = item->valueint;
        printf("查询%d历史\n", key);
        cJSON *limit_array = cJSON_GetObjectItem(data, "limit");
        long limitmin = cJSON_GetArrayItem(limit_array, 0)->valueint;
        long limitmax = cJSON_GetArrayItem(limit_array, 1)->valueint;
        // 查
        char sql_cmd[128] = {0};
        sprintf(sql_cmd, "select time,val from history where key=%d and time>=%ld and time <=%ld;", key, limitmin, limitmax);

        if (sqlite3_get_table(db, sql_cmd, &resultp, &n_row, &n_cloum, &errmsg) != SQLITE_OK)
        {
            printf("err:%s\n", errmsg);
            exit(-1);
        }
        char **p;
        p = resultp + n_cloum;
        // 序列化
        cJSON *root1 = cJSON_CreateObject();
        cJSON_AddItemToObject(root1, "type", cJSON_CreateNumber(4));
        cJSON_AddItemToObject(root1, "result", cJSON_CreateNumber(0));
        cJSON_AddItemToObject(root1, "key", cJSON_CreateNumber(key));
        cJSON *data1 = cJSON_CreateArray();
        cJSON_AddItemToObject(root1, "data", data1);
        cJSON *item1;
        printf("%d %d\n", n_row, n_cloum);
        for (i = 0; i < n_row; i++)
        {
            item1 = cJSON_CreateObject();
            for (j = 0; j < n_cloum; j++)
            {
                if (j == 0)
                {
                    double time;
                    sscanf(p[(i * n_cloum) + j], "%lf", &time);
                    cJSON_AddItemToObject(item1, "time", cJSON_CreateNumber(time));
                    printf("time:%s\n", p[(i * n_cloum) + j]);
                }
                if (j == 1)
                {
                    cJSON_AddItemToObject(item1, "val", cJSON_CreateString(p[(i * n_cloum) + j]));
                    printf("val:%s\n", p[(i * n_cloum) + j]);
                }
                // printf("值:%s\n",p[(i * n_cloum) + j]);
            }
            cJSON_AddItemToArray(data1, item1);
        }
        printf("%s\n", cJSON_Print(root1));
        publicmqtt(cJSON_Print(root1));
    }

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);

    return 1;
}

void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

void *handler_thread(void *arg)
{
    char sql_cmd[128] = {0};
    time_t current_time;

    struct shm_param shmpam;
    shm_init(&shmpam, shmemname, 1024);
    int *p = shm_getaddr(&shmpam);
    struct std_node *dev1 = (struct std_node *)(p + 1);
    struct std_node *dev = dev1;
    char val[64] = {0};
    while (1)
    {
        sleep(5);
        dev = dev1;
        for (int i = 0; i < *p; i++)
        {
            time(&current_time);
            if (dev->type == 1)
            {
                sprintf(val, "%d", dev->new_val.bool);
            }
            else if (dev->type == 2)
            {

                sprintf(val, "%d", dev->new_val.a);
            }
            else if (dev->type == 3)
            {
                sprintf(val, "%f", dev->new_val.b);
            }
            sprintf(sql_cmd, "INSERT INTO history VALUES (%d, %s, %ld );", dev->key, val, current_time);
            sqlite3_exec(db, sql_cmd, NULL, NULL, &errmsg);
            dev++;
        }
    }
    return NULL;
}

int main(int argc, char *argv[])
{
    printf("上报进程启动\n");

    // 打开数据库写

    char *errmsg;
    // 打开数据库
    int ret = sqlite3_open(sqlite3dbwhere, &db);
    if (ret != SQLITE_OK)
    {
        printf("err open:%s\n", sqlite3_errmsg(db));
        return -1;
    }
    const char *create_table_sql = "CREATE TABLE IF NOT EXISTS history ("
                                   "key INTEGER NOT NULL, "
                                   "val TEXT NOT NULL, "
                                   "time INTEGER NOT NULL, "
                                   "PRIMARY KEY (key, time));";
    ret = sqlite3_exec(db, create_table_sql, NULL, NULL, &errmsg);
    if (ret != SQLITE_OK)
    {
        printf("SQL error: %s\n", errmsg);
        sqlite3_free(errmsg); // 释放错误消息内存
    }
    else
    {
        printf("Table created successfully\n");
    }
    // 读点表确定上报方式

    FILE *fp;
    fp = fopen(pointtablewhere, "r+");

    char *buf1 = (char *)malloc(4096);

    fread(buf1, 4096, 1, fp);

    cJSON *root = cJSON_Parse(buf1);
    cJSON *report = cJSON_GetObjectItem(root, "report");

    cJSON *item = cJSON_GetObjectItem(report, "type");
    type6 = item->valueint;

    // mqtt
    char mqttaddr[64] = {0};
    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    sprintf(mqttaddr, "tcp://%s:%d", cJSON_GetObjectItem(mqtt_server, "addr")->valuestring, cJSON_GetObjectItem(mqtt_server, "port")->valueint);

    MQTTClient_create(&client, mqttaddr, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    char buf[32] = {0};
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, "down", QOS);

    // 开辟共享内存，写key值
    struct shm_param shmpam;
    shm_init(&shmpam, shmemname, 1024);
    int *p = shm_getaddr(&shmpam);
    struct std_node *dev1 = (struct std_node *)(p + 1);
    struct std_node *dev = dev1;
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    cJSON *data = cJSON_GetObjectItem(stm32, "data");
    int stm32num = cJSON_GetArraySize(data);

    for (int i = 0; i < stm32num; i++)
    {
        item = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *type = cJSON_GetObjectItem(item, "type");
        dev->key = key->valueint;
        dev->type = type->valueint;
        dev->type_dev = 0;
        dev->ret = -1;
        dev++;
    }
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    data = cJSON_GetObjectItem(modbus, "data");
    int modbusnum = cJSON_GetArraySize(data);
    for (int i = 0; i < modbusnum; i++)
    {
        item = cJSON_GetArrayItem(data, i);
        cJSON *key = cJSON_GetObjectItem(item, "key");
        cJSON *type = cJSON_GetObjectItem(item, "type");
        dev->key = key->valueint;
        dev->type = type->valueint;
        dev->type_dev = 1;
        dev->ret = -1;
        dev++;
    }
    *p = stm32num + modbusnum;
    printf("p = %d\n", *p);
    dev = dev1;

    // 开线程写历史记录
    pthread_t tid;
    if (pthread_create(&tid, NULL, handler_thread, NULL) != 0)
    {
        printf("error\n");
    }

    while (1)
    {
        // pause();

        if (type6 == 0)
        {
            // printf("type6 = %d\n", type6);
            // printf("刷新上报\n");
        }
        else if (type6 == 1) // 变化
        {
            printf("type6 = %d\n 变化上报\n", type6);
            cJSON *root = cJSON_CreateObject();
            cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
            cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
            cJSON *data = cJSON_CreateArray();
            cJSON_AddItemToObject(root, "data", data);
            cJSON *item;
            int flag = 1;
            for (int i = 0; i < *p; i++)
            {
                item = cJSON_CreateObject();
                cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(dev->key));
                char val[64] = {0};
                if (dev->type == 1)
                {
                    sprintf(val, "%d", dev->new_val.bool);
                    dev->old_val.bool = dev->new_val.bool;
                }
                else if (dev->type == 2)
                {
                    sprintf(val, "%d", dev->new_val.a);
                    dev->old_val.a = dev->new_val.a;
                }
                else if (dev->type == 3)
                {
                    sprintf(val, "%f", dev->new_val.b);
                    dev->old_val.b = dev->new_val.b;
                }
                cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
                cJSON_AddItemToArray(data, item);
                dev++;
            }

            publicmqtt(cJSON_Print(root));
            for (int i = 0; i < *p; i++)
            {
                printf("old = %.2f \n", dev[i].old_val.b);
            }

            while (type6 == 1)
            {
                sleep(2);
                // 序列化
                dev = dev1;
                cJSON *root = cJSON_CreateObject();
                cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
                cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
                cJSON *data = cJSON_CreateArray();
                cJSON_AddItemToObject(root, "data", data);
                cJSON *item;
                dev = dev1;
                for (int i = 0; i < *p; i++)
                {
                    item = cJSON_CreateObject();
                    cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(dev->key));
                    char val[64] = {0};

                    if (dev->type == 1)
                    {
                        if (dev->new_val.bool == dev->old_val.bool)
                        {
                            dev++;
                            continue;
                        }
                        else
                        {
                            sprintf(val, "%d", dev->new_val.bool);
                            dev->old_val.bool = dev->new_val.bool;
                        }
                    }
                    else if (dev->type == 2)
                    {
                        if (dev->new_val.a == dev->old_val.a)
                        {
                            dev++;
                            continue;
                        }
                        else
                        {
                            sprintf(val, "%d", dev->new_val.a);
                            dev->old_val.a = dev->new_val.a;
                        }
                    }
                    else if (dev->type == 3)
                    {
                        if (dev->new_val.b == dev->old_val.b)
                        {
                            dev++;
                            continue;
                        }
                        else
                        {
                            sprintf(val, "%f", dev->new_val.b);
                            dev->old_val.b = dev->new_val.b;
                        }
                    }
                    cJSON_AddItemToObject(item, "val", cJSON_CreateString(val));
                    cJSON_AddItemToArray(data, item);
                    dev++;
                }
                item = cJSON_GetArrayItem(data, 0);
                if (item == NULL)
                {
                    printf("无变化\n");
                    continue;
                }
                publicmqtt(cJSON_Print(root));
            }
        }
        else if (type6 == 2)
        {
            printf("type6 = %d\n", type6);
            // 周期
            item = cJSON_GetObjectItem(report, "period");
            int period = item->valueint;
            // 共享内存读
            while (type6 == 2)
            {
                sleep(period);
                // 序列化
                dev = dev1;
                cJSON *root1 = cJSON_CreateObject();
                cJSON_AddItemToObject(root1, "type", cJSON_CreateNumber(1));
                cJSON_AddItemToObject(root1, "result", cJSON_CreateNumber(0));
                cJSON *data1 = cJSON_CreateArray();
                cJSON_AddItemToObject(root1, "data", data1);
                cJSON *item1;
                for (int i = 0; i < *p; i++)
                {
                    item1 = cJSON_CreateObject();
                    cJSON_AddItemToObject(item1, "key", cJSON_CreateNumber(dev->key));
                    char val[64] = {0};
                    if (dev->type == 1)
                    {
                        sprintf(val, "%d", dev->new_val.bool);
                    }
                    else if (dev->type == 2)
                    {
                        sprintf(val, "%d", dev->new_val.a);
                    }
                    else if (dev->type == 3)
                    {
                        sprintf(val, "%f", dev->new_val.b);
                    }
                    cJSON_AddItemToObject(item1, "val", cJSON_CreateString(val));
                    cJSON_AddItemToArray(data1, item1);
                    dev++;
                }
                publicmqtt(cJSON_Print(root1));
            }
        }
    }
    sqlite3_close(db);

    printf("Waiting for up to %d seconds for publication of %s\n"
           "on topic %s for client with ClientID: %s\n",
           (int)(TIMEOUT / 1000), PAYLOAD, TOPIC, CLIENTID);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Message with delivery token %d delivered\n", token);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;

    return 0;
}
